package com.zhs;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.*;

import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.*;

@Slf4j
public class TransactionProducer {
    public static void main(String[]args) throws Exception{
        TransactionMQProducer producer=new TransactionMQProducer("transaction_producer_groupname");
        producer.setNamesrvAddr("192.168.0.241:9876");
        ExecutorService executorService=new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),(Runnable r)->{
                Thread t= new Thread(r);
                t.setName("client-transaction-msg-check-thread");
                return t;
        });
        producer.setExecutorService(executorService);
        producer.setTransactionListener(new TransactionListener() {
           @Override
           public LocalTransactionState executeLocalTransaction(Message message, Object o) {
               log.info("本地事务执行...");
               log.info("消息的标签:"+new String(message.getTags()));
               log.info("消息内容："+new String(message.getBody()));
               String tags = message.getTags();
               if("Transaction1".equals(tags)){
                    log.error("模拟本地事务执行失败");
                   return  LocalTransactionState.ROLLBACK_MESSAGE;
               }
               log.info("模拟本地事务执行成功");
               return LocalTransactionState.COMMIT_MESSAGE;
           }

           @Override
           public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
               log.info("服务器调用消息回查接口");
               log.info("消息的标签:"+new String(messageExt.getTags()));
               log.info("消息内容："+new String(messageExt.getBody()));
               return LocalTransactionState.COMMIT_MESSAGE;
           }
        });
        producer.start();
        for(int i=0;i<2;i++){
            Message message=new Message("topic_transaction","Transaction"+i,"Hello Rocket Transcation".getBytes());
            SendResult sendResult = producer.sendMessageInTransaction(message, null);
            log.info(sendResult+"");
            TimeUnit.MICROSECONDS.sleep(10);
        }
        TimeUnit.SECONDS.sleep(100);
        producer.shutdown();
    }
}
